
const { resolve, reject, some } = require('bluebird');
const kafka = require('kafka-node')
const Client = kafka.KafkaClient;
const Offset = kafka.Offset;
const Consumer = kafka.Consumer;
var utils = require('../../utils');
const deviceUtils = require('../device-utils');
function KafkaClient(_data, _logger, _events) {
    var data = _data;                   // Current webapi data
    var logger = _logger;               // Logger var working = false;                // Working flag to manage overloading polling and connection
    var working = false;                // Working flag to manage overloading polling and connection
    var connected = false;              // Connected flag
    var events = _events;               // Events to commit change to runtime
    var lastStatus = '';                // Last webapi status
    var varsValue = [];                 // Signale to send to frontend { id, type, value }
    var requestItemsMap = {};           // Map of request (JSON, CSV, XML, ...) {key: item path, value: tag}
    var overloading = 0;                // Overloading counter to mange the break connection
    var lastTimestampValue;             // Last Timestamp of asked values
    var client;
    var consumer;
    var host;
    var apiProperty = { getTags: null, postTags: null, format: 'JSON', ownFlag: true };
    this.connect = function () {
        return new Promise((resolve, reject) => {
            if (_checkConnection()) {
                try {
                    if (_checkWorking(true)) {
                        const property = getConnectionOptions(data.property)
                        console.log("连接参数",property)
                        host =  property ? property.url:''
                        if(!host){
                            reject("未获取到卡夫卡IP端口号");
                        }
                        const tags = Object.values(data.tags)
                        if (tags && tags.length) {
                            const kafkaOptions = {
                                //  自动提交配置   (false 不会提交偏移量，每次都从头开始读取)
                                autoCommit: true,
                                autoCommitIntervalMs: 5000,
                                //  如果设置为true，则consumer将从有效负载中的给定偏移量中获取消息
                                fromOffset: false,
                                timeout: 5000,
                                interval: 10000
                            };
                            const kafkaTopics = data.property.topics.map(t=>{
                                return {
                                    topic: t,//"data_iot_device_measure",
                                    partition: 0,
                                    offset: 0,
                                }
                            })
                            console.log('订阅的topics',kafkaTopics)
                            client = new Client({ kafkaHost: host });
                            consumer = new Consumer(client, kafkaTopics, kafkaOptions);
                            
                            consumer.on('message', (msg) => {
                                _emitStatus('connect-ok');
                                // var topics = Object.values(data.tags).map(t => t.address);
                                _checkWorking(false);
                                console.log('接到的消息',msg)
                                const msgValue=msg.value?JSON.parse(msg.value):{}
                                const valueList=msgValue.data || []
                                
                                for(let i=0;i<tags.length;i++){
                                    if(tags[i].deviceId===msgValue.devId){
                                        const measure =valueList.find(vv=>vv.metric===tags[i].measureCode)
                                        if(measure){
                                            const oldvalue=data.tags[tags[i].id].value
                                            measureValue=measure.value
                                            data.tags[tags[i].id].value = measureValue;
                                            data.tags[tags[i].id].timestamp = new Date().getTime();
                                            data.tags[tags[i].id].changed = oldvalue !== measureValue;
                                        }
                                    }
                                }
                                console.log("刚刚封装好的值",data.tags)
                                resolve()
                            })
                            consumer.on('error', function (error) {
                                logger.error('Error connecting to Kafka:', error);
                                reject(error)
                                
                            });
                            resolve();
                        } else {
                            reject();
                        }
                    } else {
                        reject();
                    }
                } catch (error) {
                    logger.error(`'${data.name}' try to connect error! ${err}`);
                    _checkWorking(false);
                    connected = false;
                    _emitStatus('connect-error');
                    _clearVarsValue();
                    reject();
                }
            }


        })
    }

    this.polling = function () {
        if (_checkWorking(true)) {
            if (client) {
                try {
                    var varsValueChanged = _checkVarsChanged();
                    lastTimestampValue = new Date().getTime();
                    console.log("查看当前值",varsValue)
                    _emitValues(varsValue);

                    if (this.addDaq) {
                        this.addDaq(varsValueChanged, data.name);
                    }
                } catch (err) {
                    logger.error(`'${data.name}' polling error: ${err}`);
                }
                _checkWorking(false);
            } else {
                _checkWorking(false);
            }
        }
    }

    /**
     * Return the Topics to publish that have value changed and clear value changed flag of all Topics 
     */
    var _checkVarsChanged = () => {
        const timestamp = new Date().getTime();
        var result = {};
        for (var id in data.tags) {
            if (!utils.isNullOrUndefined(data.tags[id].rawValue)) {
                data.tags[id].value = deviceUtils.tagValueCompose(data.tags[id].rawValue, data.tags[id]);
                if (this.addDaq && deviceUtils.tagDaqToSave(data.tags[id], timestamp)) {
                    result[id] = data.tags[id];
                }
            }
            data.tags[id].changed = false;
            varsValue[id] = data.tags[id];
        }
        return result;
    }

    var _checkConnection = function () {
        if (data.property.format) {
            apiProperty.getTags = data.property.address;
            apiProperty.ownFlag = false;
        }
        return true
    }

    this.setValue = function (tagId, value) {
        if (apiProperty.ownFlag && data.tags[tagId]) {
            value = _parseValue(data.tags[tagId].type, value);
            data.tags[tagId].value = deviceUtils.tagRawCalculator(value, data.tags[tagId]);
            axios.post(apiProperty.getTags, [{ id: tagId, value: data.tags[tagId].value }]).then(res => {
                lastTimestampRequest = new Date().getTime();
                logger.info(`setValue '${data.tags[tagId].name}' to ${value})`, true);
            }).catch(err => {
                logger.error(`setValue '${data.tags[tagId].name}' error! ${err}`);
            });

        } else {
            logger.error(`setValue not supported!`, true);
        }
    }

    this.getValue = function (id) {
        if (varsValue[id]) {
            return { id: id, value: varsValue[id].value, ts: lastTimestampValue };
        }
        return null;
    }


    this.disconnect = function () {
        return new Promise((resolve, reject) => {
            //关闭kafka链接
            client.close(function () {
                console.log('Kafka client connection has been closed.');
                _checkWorking(false);
                _emitStatus('connect-off');
                _clearVarsValue();
                resolve(true);
            });
        })
    }

    this.isConnected = function () {
        return true;
    }

    this.getValues = function () {
        return varsValue;
    }

    this.load = function (_data) {
        varsValue = [];
        data = JSON.parse(JSON.stringify(_data));
        try {
            requestItemsMap = {};
            var count = Object.keys(data.tags).length;
            for (var id in data.tags) {
                const address = data.tags[id].address || data.tags[id].id;
                if (!requestItemsMap[address]) {
                    requestItemsMap[address] = [data.tags[id]];
                } else {
                    requestItemsMap[address].push(data.tags[id]);
                }
            }
            logger.info(`'${data.name}' data loaded (${count})`, true);
        } catch (err) {
            logger.error(`'${data.name}' load error! ${err}`);
        }
    }

    this.bindAddDaq = function (fnc) {
        this.addDaq = fnc;
    }

    this.addDaq = null;

    this.getStatus = function () {
        return lastStatus;
    }

    //clear varsValue
    var _clearVarsValue = function () {
        for (var id in varsValue) {
            varsValue[id].value = null;
        }
        _emitValues(varsValue);
    }


    var _emitValues = function (values) {
        events.emit('device-value:changed', { id: data.id, values: values });
    }

    var _resetBrowserTimeout = function () {
        if (timeoutBrowser) {
            clearTimeout(timeoutBrowser);
        }
        timeoutBrowser = setTimeout(() => {
            if (browser && browser.connected) {
                browser.end(true);
            }
        }, 15000);
    }

    var _parseValue = function (type, value) {
        if (type === 'number') {
            return parseFloat(value);
        } else if (type === 'boolean') {
            return Boolean(value);
        } else if (type === 'string') {
            return value;
        } else {
            let val = parseFloat(value);
            if (Number.isNaN(val)) {
                // maybe boolean
                val = Number(value);
                // maybe string
                if (Number.isNaN(val)) {
                    val = value;
                }
            } else {
                val = parseFloat(val.toFixed(5));
            }
            return val;
        }
    }


    /**
 * Emit the kafka client connection status
 * @param {*} status 
 */
    var _emitStatus = function (status) {
        lastStatus = status;
        events.emit('device-status:changed', { id: data.name, status: status });
    }

    this.lastReadTimestamp = function () {
        return lastTimestampValue
    }

    var _checkWorking = function (check) {
        if (check && working) {
            overloading++;
            logger.warn(`'${data.name}' working (connection || polling) overload! ${overloading}`);
            // !The driver don't give the break connection
            if (overloading >= 3) {
                connected = false;
                // disconnect();
            } else {
                return false;
            }
        }
        working = check;
        overloading = 0;
        return true;
    }

    //获取Client 传过来的属性
    function getConnectionOptions(property) {
        return {
            url: property.address,
            format: property.format,
            keepalive: 60,
            connectTimeout: 1 * 1000,
            reconnectPeriod: 0
            //   tls: options.encryption,
            //   clientId: options.clientId,
            //   certValidation: options.certValidation,
            //   subscriptions: options.subscriptions,
            //   certificateAuthority: options.selfSignedCertificate ? options.selfSignedCertificate.data : undefined,
            //   clientCertificate: options.clientCertificate ? options.clientCertificate.data : undefined,
            //   clientKey: options.clientKey ? options.clientKey.data : undefined,
        }
    }
}
module.exports = {
    init: function (settings) {
        // deviceCloseTimeout = settings.deviceCloseTimeout || 15000;
    },
    create: function (data, logger, events) {
        return new KafkaClient(data, logger, events);
    },
}
